package com.alibaba.nacos.client.naming.core;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.listener.EventListener;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
import com.alibaba.nacos.client.monitor.MetricsMonitor;
import com.alibaba.nacos.client.naming.backups.FailoverReactor;
import com.alibaba.nacos.client.naming.beat.BeatInfo;
import com.alibaba.nacos.client.naming.beat.BeatReactor;
import com.alibaba.nacos.client.naming.cache.DiskCache;
import com.alibaba.nacos.client.naming.event.InstancesChangeEvent;
import com.alibaba.nacos.client.naming.event.InstancesChangeNotifier;
import com.alibaba.nacos.client.naming.net.NamingProxy;
import com.alibaba.nacos.client.naming.utils.CollectionUtils;
import com.alibaba.nacos.client.naming.utils.UtilAndComs;
import com.alibaba.nacos.client.utils.LogUtils;
import com.alibaba.nacos.common.lifecycle.Closeable;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.common.utils.JacksonUtils;
import com.alibaba.nacos.common.utils.StringUtils;
import com.alibaba.nacos.common.utils.ThreadUtils;

import java.util.*;
import java.util.concurrent.*;

/**
 * @ClassName HostReactor
 * @Description: TODO
 * @Author linjiahao
 * @Date 2023/2/1
 **/
public class HostReactor implements Closeable {
    private static final long DEFAULT_DELAY = 1000L;
    private static final long UPDATE_HOLD_INTERVAL = 5000L;
    private final Map<String, ScheduledFuture<?>> futureMap;
    private final Map<String, ServiceInfo> serviceInfoMap;
    private final Map<String, Object> updatingMap;
    private final PushReceiver pushReceiver;
    private final BeatReactor beatReactor;
    private final NamingProxy serverProxy;
    private final FailoverReactor failoverReactor;
    private final String cacheDir;
    private final boolean pushEmptyProtection;
    private final ScheduledExecutorService executor;
    private final InstancesChangeNotifier notifier;

    public HostReactor(NamingProxy serverProxy, BeatReactor beatReactor, String cacheDir) {
        this(serverProxy, beatReactor, cacheDir, false, false, UtilAndComs.DEFAULT_POLLING_THREAD_COUNT);
    }

    public HostReactor(NamingProxy serverProxy, BeatReactor beatReactor, String cacheDir, boolean loadCacheAtStart, boolean pushEmptyProtection, int pollingThreadCount) {
        this.futureMap = new HashMap();
        this.executor = new ScheduledThreadPoolExecutor(pollingThreadCount, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("com.alibaba.nacos.client.naming.updater");
                return thread;
            }
        });
        this.beatReactor = beatReactor;
        this.serverProxy = serverProxy;
        this.cacheDir = cacheDir;
        if (loadCacheAtStart) {
            this.serviceInfoMap = new ConcurrentHashMap(DiskCache.read(this.cacheDir));
        } else {
            this.serviceInfoMap = new ConcurrentHashMap(16);
        }

        this.pushEmptyProtection = pushEmptyProtection;
        this.updatingMap = new ConcurrentHashMap();
        this.failoverReactor = new FailoverReactor(this, cacheDir);
        this.pushReceiver = new PushReceiver(this);
        this.notifier = new InstancesChangeNotifier();
        NotifyCenter.registerToPublisher(InstancesChangeEvent.class, 16384);
        NotifyCenter.registerSubscriber(this.notifier);
    }

    public Map<String, ServiceInfo> getServiceInfoMap() {
        return this.serviceInfoMap;
    }

    public synchronized ScheduledFuture<?> addTask(HostReactor.UpdateTask task) {
        return this.executor.schedule(task, 1000L, TimeUnit.MILLISECONDS);
    }

    public void subscribe(String serviceName, String clusters, EventListener eventListener) {
        this.notifier.registerListener(serviceName, clusters, eventListener);
        this.getServiceInfo(serviceName, clusters);
    }

    public void unSubscribe(String serviceName, String clusters, EventListener eventListener) {
        this.notifier.deregisterListener(serviceName, clusters, eventListener);
    }

    public List<ServiceInfo> getSubscribeServices() {
        return this.notifier.getSubscribeServices();
    }

    public ServiceInfo processServiceJson(String json) {
        ServiceInfo serviceInfo = (ServiceInfo) JacksonUtils.toObj(json, ServiceInfo.class);
        ServiceInfo oldService = (ServiceInfo)this.serviceInfoMap.get(serviceInfo.getKey());
        Integer eventType = 0;
        if (this.pushEmptyProtection && !serviceInfo.validate()) {
            return oldService;
        } else {
            boolean changed = false;
            if (oldService != null) {
                if (oldService.getLastRefTime() > serviceInfo.getLastRefTime()) {
                    LogUtils.NAMING_LOGGER.warn("out of date data received, old-t: " + oldService.getLastRefTime() + ", new-t: " + serviceInfo.getLastRefTime());
                }

                this.serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
                Map<String, Instance> oldHostMap = new HashMap(oldService.getHosts().size());
                Iterator var6 = oldService.getHosts().iterator();

                while(var6.hasNext()) {
                    Instance host = (Instance)var6.next();
                    oldHostMap.put(host.toInetAddr(), host);
                }

                Map<String, Instance> newHostMap = new HashMap(serviceInfo.getHosts().size());
                Iterator var16 = serviceInfo.getHosts().iterator();

                while(var16.hasNext()) {
                    Instance host = (Instance)var16.next();
                    newHostMap.put(host.toInetAddr(), host);
                }

                Set<Instance> modHosts = new HashSet();
                Set<Instance> newHosts = new HashSet();
                Set<Instance> remvHosts = new HashSet();
                List<Map.Entry<String, Instance>> newServiceHosts = new ArrayList(newHostMap.entrySet());
                Iterator var11 = newServiceHosts.iterator();

                while(true) {
                    Map.Entry entry;
                    Instance host;
                    String key;
                    while(var11.hasNext()) {
                        entry = (Map.Entry)var11.next();
                        host = (Instance)entry.getValue();
                        key = (String)entry.getKey();
                        if (oldHostMap.containsKey(key) && !StringUtils.equals(host.toString(), ((Instance)oldHostMap.get(key)).toString())) {
                            modHosts.add(host);
                        } else if (!oldHostMap.containsKey(key)) {
                            newHosts.add(host);
                        }
                    }

                    var11 = oldHostMap.entrySet().iterator();

                    while(var11.hasNext()) {
                        entry = (Map.Entry)var11.next();
                        host = (Instance)entry.getValue();
                        key = (String)entry.getKey();
                        if (!newHostMap.containsKey(key) && !newHostMap.containsKey(key)) {
                            remvHosts.add(host);
                        }
                    }
                    if (newHosts.size() > 0) {
                        eventType = 2;
                        changed = true;
                        LogUtils.NAMING_LOGGER.info("new ips(" + newHosts.size() + ") service: " + serviceInfo.getKey() + " -> " + JacksonUtils.toJson(newHosts));
                    }

                    if (remvHosts.size() > 0) {
                        eventType = 3;
                        changed = true;
                        LogUtils.NAMING_LOGGER.info("removed ips(" + remvHosts.size() + ") service: " + serviceInfo.getKey() + " -> " + JacksonUtils.toJson(remvHosts));
                    }

                    if (modHosts.size() > 0) {
                        eventType = 4;
                        changed = true;
                        this.updateBeatInfo(modHosts);
                        LogUtils.NAMING_LOGGER.info("modified ips(" + modHosts.size() + ") service: " + serviceInfo.getKey() + " -> " + JacksonUtils.toJson(modHosts));
                    }

                    serviceInfo.setJsonFromServer(json);
                    if (newHosts.size() > 0 || remvHosts.size() > 0 || modHosts.size() > 0) {
                        NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(), serviceInfo.getClusters(), serviceInfo.getHosts(),eventType));
                        DiskCache.write(serviceInfo, this.cacheDir);
                    }
                    break;
                }
            } else {
                changed = true;
                LogUtils.NAMING_LOGGER.info("init new ips(" + serviceInfo.ipCount() + ") service: " + serviceInfo.getKey() + " -> " + JacksonUtils.toJson(serviceInfo.getHosts()));
                this.serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
                eventType = 1;
                NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(), serviceInfo.getClusters(), serviceInfo.getHosts(),eventType));
                serviceInfo.setJsonFromServer(json);
                DiskCache.write(serviceInfo, this.cacheDir);
            }

            MetricsMonitor.getServiceInfoMapSizeMonitor().set((double)this.serviceInfoMap.size());
            if (changed) {
                LogUtils.NAMING_LOGGER.info("current ips:(" + serviceInfo.ipCount() + ") service: " + serviceInfo.getKey() + " -> " + JacksonUtils.toJson(serviceInfo.getHosts()));
            }

            return serviceInfo;
        }
    }

    private void updateBeatInfo(Set<Instance> modHosts) {
        Iterator var2 = modHosts.iterator();

        while(var2.hasNext()) {
            Instance instance = (Instance)var2.next();
            String key = this.beatReactor.buildKey(instance.getServiceName(), instance.getIp(), instance.getPort());
            if (this.beatReactor.dom2Beat.containsKey(key) && instance.isEphemeral()) {
                BeatInfo beatInfo = this.beatReactor.buildBeatInfo(instance);
                this.beatReactor.addBeatInfo(instance.getServiceName(), beatInfo);
            }
        }

    }

    private ServiceInfo getServiceInfo0(String serviceName, String clusters) {
        String key = ServiceInfo.getKey(serviceName, clusters);
        return (ServiceInfo)this.serviceInfoMap.get(key);
    }

    public ServiceInfo getServiceInfoDirectlyFromServer(String serviceName, String clusters) throws NacosException {
        String result = this.serverProxy.queryList(serviceName, clusters, 0, false);
        return StringUtils.isNotEmpty(result) ? (ServiceInfo)JacksonUtils.toObj(result, ServiceInfo.class) : null;
    }

    public ServiceInfo getServiceInfo(String serviceName, String clusters) {
        LogUtils.NAMING_LOGGER.debug("failover-mode: " + this.failoverReactor.isFailoverSwitch());
        String key = ServiceInfo.getKey(serviceName, clusters);
        if (this.failoverReactor.isFailoverSwitch()) {
            return this.failoverReactor.getService(key);
        } else {
            ServiceInfo serviceObj = this.getServiceInfo0(serviceName, clusters);
            if (null == serviceObj) {
                serviceObj = new ServiceInfo(serviceName, clusters);
                this.serviceInfoMap.put(serviceObj.getKey(), serviceObj);
                this.updatingMap.put(serviceName, new Object());
                this.updateServiceNow(serviceName, clusters);
                this.updatingMap.remove(serviceName);
            } else if (this.updatingMap.containsKey(serviceName)) {
                synchronized(serviceObj) {
                    try {
                        serviceObj.wait(5000L);
                    } catch (InterruptedException var8) {
                        LogUtils.NAMING_LOGGER.error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, var8);
                    }
                }
            }

            this.scheduleUpdateIfAbsent(serviceName, clusters);
            return (ServiceInfo)this.serviceInfoMap.get(serviceObj.getKey());
        }
    }

    private void updateServiceNow(String serviceName, String clusters) {
        try {
            this.updateService(serviceName, clusters);
        } catch (NacosException var4) {
            LogUtils.NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, var4);
        }

    }

    public void scheduleUpdateIfAbsent(String serviceName, String clusters) {
        if (this.futureMap.get(ServiceInfo.getKey(serviceName, clusters)) == null) {
            synchronized(this.futureMap) {
                if (this.futureMap.get(ServiceInfo.getKey(serviceName, clusters)) == null) {
                    ScheduledFuture<?> future = this.addTask(new HostReactor.UpdateTask(serviceName, clusters));
                    this.futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);
                }
            }
        }
    }

    public void updateService(String serviceName, String clusters) throws NacosException {
        ServiceInfo oldService = this.getServiceInfo0(serviceName, clusters);
        boolean var12 = false;

        try {
            var12 = true;
            String result = this.serverProxy.queryList(serviceName, clusters, this.pushReceiver.getUdpPort(), false);
            if (StringUtils.isNotEmpty(result)) {
                this.processServiceJson(result);
                var12 = false;
            } else {
                var12 = false;
            }
        } finally {
            if (var12) {
                if (oldService != null) {
                    synchronized(oldService) {
                        oldService.notifyAll();
                    }
                }

            }
        }

        if (oldService != null) {
            synchronized(oldService) {
                oldService.notifyAll();
            }
        }

    }

    public void refreshOnly(String serviceName, String clusters) {
        try {
            this.serverProxy.queryList(serviceName, clusters, this.pushReceiver.getUdpPort(), false);
        } catch (Exception var4) {
            LogUtils.NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, var4);
        }

    }

    @Override
    public void shutdown() throws NacosException {
        String className = this.getClass().getName();
        LogUtils.NAMING_LOGGER.info("{} do shutdown begin", className);
        ThreadUtils.shutdownThreadPool(this.executor, LogUtils.NAMING_LOGGER);
        this.pushReceiver.shutdown();
        this.failoverReactor.shutdown();
        NotifyCenter.deregisterSubscriber(this.notifier);
        LogUtils.NAMING_LOGGER.info("{} do shutdown stop", className);
    }

    public class UpdateTask implements Runnable {
        long lastRefTime = 9223372036854775807L;
        private final String clusters;
        private final String serviceName;
        private int failCount = 0;

        public UpdateTask(String serviceName, String clusters) {
            this.serviceName = serviceName;
            this.clusters = clusters;
        }

        private void incFailCount() {
            int limit = 6;
            if (this.failCount != limit) {
                ++this.failCount;
            }
        }

        private void resetFailCount() {
            this.failCount = 0;
        }

        @Override
        public void run() {
            long delayTime = 1000L;

            try {
                ServiceInfo serviceObj = (ServiceInfo)HostReactor.this.serviceInfoMap.get(ServiceInfo.getKey(this.serviceName, this.clusters));
                if (serviceObj == null) {
                    HostReactor.this.updateService(this.serviceName, this.clusters);
                    return;
                }

                if (serviceObj.getLastRefTime() <= this.lastRefTime) {
                    HostReactor.this.updateService(this.serviceName, this.clusters);
                    serviceObj = (ServiceInfo)HostReactor.this.serviceInfoMap.get(ServiceInfo.getKey(this.serviceName, this.clusters));
                } else {
                    HostReactor.this.refreshOnly(this.serviceName, this.clusters);
                }

                this.lastRefTime = serviceObj.getLastRefTime();
                if (!HostReactor.this.notifier.isSubscribed(this.serviceName, this.clusters) && !HostReactor.this.futureMap.containsKey(ServiceInfo.getKey(this.serviceName, this.clusters))) {
                    LogUtils.NAMING_LOGGER.info("update task is stopped, service:" + this.serviceName + ", clusters:" + this.clusters);
                    return;
                }

                if (CollectionUtils.isEmpty(serviceObj.getHosts())) {
                    this.incFailCount();
                    return;
                }

                delayTime = serviceObj.getCacheMillis();
                this.resetFailCount();
            } catch (Throwable var7) {
                this.incFailCount();
                LogUtils.NAMING_LOGGER.warn("[NA] failed to update serviceName: " + this.serviceName, var7);
            } finally {
                HostReactor.this.executor.schedule(this, Math.min(delayTime << this.failCount, 60000L), TimeUnit.MILLISECONDS);
            }

        }
    }
}
